package com.apps.sdses.flink141.demo;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;

import java.io.Serializable;
import java.util.LinkedList;
import java.util.List;

/**
 * @author jiwei
 * @description
 * @date 2023/8/24 16:29
 */
public class FieldsToJsonDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        List<Column> columns = new LinkedList<>();
        columns.add(new Column("id", DataTypes.BIGINT()));
        columns.add(new Column("sid", DataTypes.STRING()));
        columns.add(new Column("ts", DataTypes.TIMESTAMP(3)));
        columns.add(new Column("flag", DataTypes.BOOLEAN()));
        columns.add(new Column("vc", DataTypes.INT()));

        Schema.Builder schemaBuilder = Schema.newBuilder();
        for (Column column : columns) {
            schemaBuilder = schemaBuilder.column(column.getName(), column.getDataType());
        }
        Schema schema = schemaBuilder.watermark("ts", "ts - INTERVAL '3' SECOND")
                .build();

        Table table = tableEnv.from(
                TableDescriptor.forConnector("datagen")
                        .schema(schema)
//                        .option("rows-per-second", "1")
                        .option("number-of-rows", "10")
                        .option("fields.sid.length", "5")
                        .option("fields.id.min", "0")
                        .option("fields.id.max", "10")
                        .option("fields.vc.min", "50")
                        .option("fields.vc.max", "100")
                        .build()
        );

        List<Column> fieldsToJson = new LinkedList<>();
        fieldsToJson.add(new Column("id", DataTypes.BIGINT()));
        fieldsToJson.add(new Column("sid", DataTypes.STRING()));
        fieldsToJson.add(new Column("ts", DataTypes.TIMESTAMP(3)));
        fieldsToJson.add(new Column("flag", DataTypes.BOOLEAN()));


        //多个字段转成一个json字符串
        DataStream<Row> dataStream = tableEnv.toDataStream(table);
        SingleOutputStreamOperator<Row> mapStream = dataStream.map(new FieldsToJsonMapFunction(fieldsToJson)).returns(Types.ROW_NAMED(new String[]{"new_field", "a"}, Types.STRING, Types.STRING));

        mapStream.print();

        env.execute();

    }
    @Data
    @AllArgsConstructor
    static class Column implements Serializable {
        public String name;
        public DataType dataType;
//        public Object oriType;
    }

}
